package ru.nagakhl.whirlpool;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.logging.Level;
import java.util.logging.Logger;

import org.apache.cassandra.thrift.Cassandra;
import org.apache.cassandra.thrift.Column;
import org.apache.cassandra.thrift.ColumnOrSuperColumn;
import org.apache.cassandra.thrift.InvalidRequestException;
import org.apache.cassandra.thrift.TimedOutException;
import org.apache.cassandra.thrift.UnavailableException;
import org.apache.cassandra.thrift.Cassandra.Client;
import org.apache.thrift.TException;
import org.apache.thrift.transport.TTransportException;

import ru.nagakhl.whirlpool.exceptions.ConnectionClosedException;
import ru.nagakhl.whirlpool.exceptions.ConnectorException;

public class ConnectionImpl implements Connection {

	protected boolean closed = false;

	protected Cassandra.Client client;

	protected ManagerConnector connector;

	protected Logger log = Logger.getLogger(getClass().getSimpleName());

	public ConnectionImpl(Cassandra.Client client, ManagerConnector r) {
		this.client = client;
		this.connector = r;
	}

	public void close() {
		flush();
		connector.releaseClient(client);
		// will be better to ensure that after releasing connection
		// no more operations on it will be performed
		client = null;
		closed = true;
	}

	public void delete(Object id) {
		checkClosed();
	}

	public void flush() {
		checkClosed();
		try {
			client.getOutputProtocol().getTransport().flush();
		} catch (TTransportException e) {
			log.log(Level.SEVERE, "Cannot flush data!", e);
			throw new ConnectorException("Cannot flush data!", e);
		}
	}

	public <T> T get(Class<T> cls, Object id) {
		checkClosed();
		return null;
	}

	public <T> List<T> getAll(Class<T> cls) {
		checkClosed();
		return null;
	}

	public Client getClient() {
		checkClosed();
		return null;
	}

	public void merge(Object element) {
		checkClosed();

	}

	public void save(Object element) {
		checkClosed();
		long timestamp = System.currentTimeMillis();
		Metadata m = connector.getMetadata(element.getClass());
		Map<String, List<ColumnOrSuperColumn>> data = new HashMap<String, List<ColumnOrSuperColumn>>();
		List<ColumnOrSuperColumn> columns = new ArrayList<ColumnOrSuperColumn>(m.fieldCount());
		Map<byte[], byte[]> fields = m.getFields(element);
		for (byte[] key : fields.keySet()) {
			ColumnOrSuperColumn column = new ColumnOrSuperColumn();
			if (!m.isSuperColumn()) {
				column.setColumn(new Column(key, fields.get(key), timestamp));
			}
			columns.add(column);
		}
		data.put(m.getColumnFamily(), columns);
		//TODO: add exception processing here
		try {
			client.batch_insert(m.getKeyspace(), m.getId(element), data, m.getLevel());
		} catch (InvalidRequestException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (UnavailableException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (TimedOutException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (TException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	protected void checkClosed() {
		if (closed)
			throw new ConnectionClosedException();
	}
}
